package com.itcast.flink.usage.stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @program: flink-app
 * @description: 流处理案例
 * @author: zhanghz001
 * @create: 2021-07-22 10:41
 **/
public class ZhzStreamProcessorApplication {
    public static void main(String[] args) throws Exception {
        //创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //读取socket数据源
        DataStreamSource<String> socketStr = env.socketTextStream("192.168.23.140", 9911, "\n");
        //    转换处理流数据
        
        socketStr.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value,
                                Collector<Tuple2<String, Integer>> collector)
                    throws Exception {
                String[] arrValue = value.split("\t");
                collector.collect(new Tuple2<String, Integer>(arrValue[0], 1));
            }
        }).keyBy(0).timeWindow(Time.seconds(5))
                .sum(1).print().setParallelism(2);
        env.execute("accessLog");
    }
}
